package com.xdf.ecommerce.service.hystrix;

import com.alibaba.fastjson.JSON;
import com.google.common.eventbus.Subscribe;
import com.netflix.hystrix.HystrixCommandGroupKey;
import com.netflix.hystrix.HystrixCommandKey;
import com.netflix.hystrix.HystrixCommandProperties;
import com.netflix.hystrix.HystrixObservableCommand;
import com.xdf.ecommerce.service.NacosClientService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.client.ServiceInstance;

import java.util.Collections;
import java.util.List;
import rx.Observable;
import rx.Subscriber;

/**
 * 使用信号量的方式实现舱壁隔离策略，保护方法 NacosClientService.getNacosClientInfo
 * @author chanchaw
 * @create 2025-10-30 12:21
 */
@Slf4j
public class NacosClientHystrixObservableCommand extends HystrixObservableCommand<List<ServiceInstance>> {
    private NacosClientService nacosClientService;
    private List<String> serviceIds;
    public NacosClientHystrixObservableCommand(NacosClientService nacosClientService, List<String> serviceIds){
        super(
                HystrixObservableCommand.Setter
                        .withGroupKey(HystrixCommandGroupKey
                                .Factory.asKey("NacosClientService"))
                        .andCommandKey(HystrixCommandKey
                                .Factory.asKey("NacosClientHystrixObservableCommand"))
                        .andCommandPropertiesDefaults(
                                HystrixCommandProperties.Setter()
                                        .withFallbackEnabled(true)          // 开启降级
                                        .withCircuitBreakerEnabled(true)    // 开启熔断器
                        )
        );
        this.nacosClientService = nacosClientService;
        this.serviceIds = serviceIds;
    }

    /**
     * 要保护的方法调用写在这里
     * @return
     */
    @Override
    protected Observable<List<ServiceInstance>> construct(){
        // Observable 有三个关键的回调函数：onNext, onCompleted, onError
        // 分别表示：开始执行，已完成，出错
        return Observable.create(new Observable.OnSubscribe<List<ServiceInstance>>(){
            @Override
            public void call(Subscriber<? super List<ServiceInstance>> subscriber) {
                try {
                    if(!subscriber.isUnsubscribed()){
                        log.info("测试 Hystrix 信号量舱壁隔离，订阅命令任务，服务ID：[{}]，线程名称：[{}]", JSON.toJSONString(serviceIds), Thread.currentThread().getName());
                        serviceIds.forEach(item -> subscriber.onNext(nacosClientService.getNacosClientInfo(item)));
                        subscriber.onCompleted();
                        log.info("信号量舱壁隔离查询获取服务实例完成，服务ID：[{}]，线程名称：[{}]", JSON.toJSONString(serviceIds), Thread.currentThread().getName());
                    }
                } catch(Exception ex){
                    subscriber.onError(ex);
                }
            }
        });
    }

    /**
     * 服务降级
     * @return
     */
    @Override
    protected Observable<List<ServiceInstance>> resumeWithFallback() {
        return Observable.create(new Observable.OnSubscribe<List<ServiceInstance>>(){
            @Override
            public void call(Subscriber<? super List<ServiceInstance>> subscriber) {
                try {
                    if(!subscriber.isUnsubscribed()){
                        log.info("[信号量舱壁模式 fallBack] 测试 Hystrix 信号量舱壁隔离，订阅命令任务，服务ID：[{}]，线程名称：[{}]", JSON.toJSONString(serviceIds), Thread.currentThread().getName());
                        // 服务降级时返回空数组
                        subscriber.onNext(Collections.emptyList());
                        subscriber.onCompleted();
                        log.info("[信号量舱壁模式 fallBack] 信号量舱壁隔离查询获取服务实例完成，服务ID：[{}]，线程名称：[{}]", JSON.toJSONString(serviceIds), Thread.currentThread().getName());
                    }
                } catch(Exception ex){
                    subscriber.onError(ex);
                }
            }
        });
    }
}
